package com.cuijian.demo.service;

import com.cuijian.demo.config.RocketMQConfig;
import com.cuijian.demo.mq.MQConstants;
import com.cuijian.demo.model.LogisticsOrder;
import com.cuijian.demo.model.User;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import com.alibaba.fastjson2.JSON;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.concurrent.atomic.AtomicBoolean;

public class RocketMQConsumerService {

    private DefaultMQPushConsumer consumer;
    private RocketMQProducerService producerService;
    private AtomicBoolean running = new AtomicBoolean(false);
    private static RocketMQConsumerService instance;

    private RocketMQConsumerService() {
        this.producerService = new RocketMQProducerService();
        initConsumer();
    }

    public static synchronized RocketMQConsumerService getInstance() {
        if (instance == null) {
            instance = new RocketMQConsumerService();
        }
        return instance;
    }

    /**
     * 初始化消费者并订阅消息
     */
    private void initConsumer() {
        if (!RocketMQConfig.getInstance().isMqEnabled()) {
            System.out.println("⚠️ MQ未启用，跳过消费者初始化");
            return;
        }

        try {
            consumer = new DefaultMQPushConsumer(RocketMQConfig.getInstance().getConsumerGroup());
            consumer.setNamesrvAddr(RocketMQConfig.getInstance().getNameServer());
            consumer.setConsumeFromWhere(org.apache.rocketmq.common.consumer.ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            // 设置消费线程数
            consumer.setConsumeThreadMin(5);
            consumer.setConsumeThreadMax(10);

            // 订阅主题和标签
            subscribeTopics();

            // 注册消息监听器
            consumer.registerMessageListener(createMessageListener());

            // 启动消费者
            consumer.start();
            running.set(true);

            System.out.println("✅ RocketMQ Consumer启动成功");
            System.out.println("🎯 Consumer Group: " + RocketMQConfig.getInstance().getConsumerGroup());
            System.out.println("📡 订阅主题: " + getSubscribedTopics());

        } catch (Exception e) {
            running.set(false);
            System.err.println("❌ RocketMQ Consumer启动失败: " + e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 订阅主题
     */
    private void subscribeTopics() {
        try {
            // 订阅订单主题 - 所有标签
            consumer.subscribe(MQConstants.TOPIC_ORDER, "*");
            System.out.println("📨 订阅订单主题: " + MQConstants.TOPIC_ORDER);

            // 订阅用户主题 - 所有标签
            consumer.subscribe(MQConstants.TOPIC_USER, "*");
            System.out.println("👥 订阅用户主题: " + MQConstants.TOPIC_USER);

            // 订阅通知主题 - 所有标签
            consumer.subscribe(MQConstants.TOPIC_NOTIFICATION, "*");
            System.out.println("🔔 订阅通知主题: " + MQConstants.TOPIC_NOTIFICATION);

        } catch (Exception e) {
            System.err.println("❌ 订阅主题失败: " + e.getMessage());
            throw new RuntimeException("订阅主题失败", e);
        }
    }

    /**
     * 创建消息监听器
     */
    private MessageListenerConcurrently createMessageListener() {
        return (msgs, context) -> {
            for (MessageExt msg : msgs) {
                try {
                    boolean processSuccess = processSingleMessage(msg);
                    if (!processSuccess) {
                        System.err.println("❌ 消息处理失败，稍后重试: " + msg.getMsgId());
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                } catch (Exception e) {
                    System.err.println("❌ 消息处理异常: " + e.getMessage());
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        };
    }

    /**
     * 处理单条消息
     */
    private boolean processSingleMessage(MessageExt msg) {
        String topic = msg.getTopic();
        String tags = msg.getTags();
        String body = new String(msg.getBody(), StandardCharsets.UTF_8);

        System.out.printf("📩 收到消息: Topic=%s, Tags=%s, MsgId=%s%n",
                topic, tags, msg.getMsgId());
        System.out.printf("📦 消息内容: %s%n", body);

        try {
            switch (topic) {
                case MQConstants.TOPIC_ORDER:
                    return processOrderMessage(tags, body, msg.getMsgId());
                case MQConstants.TOPIC_USER:
                    return processUserMessage(tags, body, msg.getMsgId());
                case MQConstants.TOPIC_NOTIFICATION:
                    return processNotificationMessage(tags, body, msg.getMsgId());
                default:
                    System.out.println("⚠️ 未知主题，跳过处理: " + topic);
                    return true; // 未知主题也返回成功，避免重复消费
            }
        } catch (Exception e) {
            System.err.printf("❌ 处理消息失败: MsgId=%s, 错误=%s%n", msg.getMsgId(), e.getMessage());
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 处理订单相关消息
     */
    private boolean processOrderMessage(String tags, String body, String msgId) {
        System.out.printf("📦 处理订单消息: Tags=%s, MsgId=%s%n", tags, msgId);

        try {
            switch (tags) {
                case MQConstants.TAG_ORDER_CREATED:
                    return processOrderCreated(body, msgId);
                case MQConstants.TAG_ORDER_STATUS_CHANGED:
                    return processOrderStatusChanged(body, msgId);
                case MQConstants.TAG_ORDER_DELIVERED:
                    return processOrderDelivered(body, msgId);
                case MQConstants.TAG_ORDER_UPDATED:
                    return processOrderUpdated(body, msgId);
                default:
                    System.out.println("⚠️ 未知订单标签，跳过处理: " + tags);
                    return true;
            }
        } catch (Exception e) {
            System.err.printf("❌ 处理订单消息失败: Tags=%s, MsgId=%s, 错误=%s%n",
                    tags, msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理订单创建消息
     */
    private boolean processOrderCreated(String body, String msgId) {
        try {
            RocketMQProducerService.OrderCreatedMessage message =
                    JSON.parseObject(body, RocketMQProducerService.OrderCreatedMessage.class);

            System.out.printf("✅ 处理订单创建: 订单ID=%d, 物流单号=%s, 收件人=%s%n",
                    message.getOrderId(), message.getTrackingNumber(), message.getRecipientName());

            // 1. 发送创建通知给管理员
            sendOrderCreatedNotification(message);

            // 2. 更新相关统计信息
            updateOrderStatistics(message);

            // 3. 记录操作日志
            logOrderCreation(message);

            System.out.printf("✅ 订单创建消息处理完成: MsgId=%s%n", msgId);
            return true;

        } catch (Exception e) {
            System.err.printf("❌ 处理订单创建消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理订单状态变更消息
     */
    private boolean processOrderStatusChanged(String body, String msgId) {
        try {
            RocketMQProducerService.OrderStatusChangeMessage message =
                    JSON.parseObject(body, RocketMQProducerService.OrderStatusChangeMessage.class);

            System.out.printf("🔄 处理订单状态变更: 订单ID=%d, 状态 %d→%d%n",
                    message.getOrderId(), message.getOldStatus(), message.getNewStatus());

            // 1. 发送状态变更通知
            sendStatusChangeNotification(message);

            // 2. 更新订单时间戳
            updateOrderTimestamp(message.getOrderId());

            System.out.printf("✅ 订单状态变更消息处理完成: MsgId=%s%n", msgId);
            return true;

        } catch (Exception e) {
            System.err.printf("❌ 处理订单状态变更消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理订单签收消息
     */
    private boolean processOrderDelivered(String body, String msgId) {
        try {
            RocketMQProducerService.OrderDeliveredMessage message =
                    JSON.parseObject(body, RocketMQProducerService.OrderDeliveredMessage.class);

            System.out.printf("🎉 处理订单签收: 订单ID=%d, 物流单号=%s%n",
                    message.getOrderId(), message.getTrackingNumber());

            // 1. 发送签收成功通知
            sendDeliverySuccessNotification(message);

            // 2. 更新订单完成时间
            markOrderAsCompleted(message.getOrderId());

            System.out.printf("✅ 订单签收消息处理完成: MsgId=%s%n", msgId);
            return true;

        } catch (Exception e) {
            System.err.printf("❌ 处理订单签收消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理订单更新消息
     */
    private boolean processOrderUpdated(String body, String msgId) {
        try {
            System.out.printf("📝 处理订单更新消息: MsgId=%s%n", msgId);
            // 这里可以添加订单更新的业务逻辑
            return true;
        } catch (Exception e) {
            System.err.printf("❌ 处理订单更新消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理用户相关消息
     */
    private boolean processUserMessage(String tags, String body, String msgId) {
        System.out.printf("👤 处理用户消息: Tags=%s, MsgId=%s%n", tags, msgId);

        try {
            switch (tags) {
                case MQConstants.TAG_USER_REGISTER:
                    return processUserRegistered(body, msgId);
                case MQConstants.TAG_USER_LOGIN:
                    return processUserLogin(body, msgId);
                case MQConstants.TAG_USER_UPDATE:
                    return processUserUpdated(body, msgId);
                default:
                    System.out.println("⚠️ 未知用户标签，跳过处理: " + tags);
                    return true;
            }
        } catch (Exception e) {
            System.err.printf("❌ 处理用户消息失败: Tags=%s, MsgId=%s, 错误=%s%n",
                    tags, msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理用户注册消息
     */
    private boolean processUserRegistered(String body, String msgId) {
        try {
            RocketMQProducerService.UserRegisterMessage message =
                    JSON.parseObject(body, RocketMQProducerService.UserRegisterMessage.class);

            System.out.printf("🎊 处理用户注册: 用户ID=%d, 用户名=%s, 邮箱=%s%n",
                    message.getUserId(), message.getUsername(), message.getEmail());

            // 1. 发送欢迎邮件
            sendWelcomeEmail(message);

            // 2. 发送系统通知
            sendNewUserNotification(message);

            // 3. 初始化用户统计
            initUserStatistics(message.getUserId());

            System.out.printf("✅ 用户注册消息处理完成: MsgId=%s%n", msgId);
            return true;

        } catch (Exception e) {
            System.err.printf("❌ 处理用户注册消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理用户登录消息
     */
    private boolean processUserLogin(String body, String msgId) {
        try {
            System.out.printf("🔐 处理用户登录消息: MsgId=%s%n", msgId);
            // 这里可以添加用户登录的业务逻辑，如记录登录日志等
            return true;
        } catch (Exception e) {
            System.err.printf("❌ 处理用户登录消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理用户更新消息
     */
    private boolean processUserUpdated(String body, String msgId) {
        try {
            System.out.printf("✏️ 处理用户更新消息: MsgId=%s%n", msgId);
            // 这里可以添加用户信息更新的业务逻辑
            return true;
        } catch (Exception e) {
            System.err.printf("❌ 处理用户更新消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理通知消息
     */
    private boolean processNotificationMessage(String tags, String body, String msgId) {
        try {
            RocketMQProducerService.NotificationMessage message =
                    JSON.parseObject(body, RocketMQProducerService.NotificationMessage.class);

            System.out.printf("🔔 处理通知消息: 类型=%s, 接收方=%s, 标题=%s%n",
                    message.getType(), message.getTo(), message.getTitle());

            // 根据类型处理不同的通知
            switch (message.getType().toLowerCase()) {
                case "email":
                    return processEmailNotification(message, msgId);
                case "sms":
                    return processSmsNotification(message, msgId);
                case "system":
                    return processSystemNotification(message, msgId);
                default:
                    System.out.println("⚠️ 未知通知类型: " + message.getType());
                    return true;
            }

        } catch (Exception e) {
            System.err.printf("❌ 处理通知消息失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理邮件通知
     */
    private boolean processEmailNotification(RocketMQProducerService.NotificationMessage message, String msgId) {
        try {
            // 模拟发送邮件
            System.out.printf("📧 发送邮件通知: 给%s, 标题: %s%n", message.getTo(), message.getTitle());
            System.out.printf("📝 邮件内容: %s%n", message.getContent());

            // 这里可以集成真实的邮件发送服务
            // mailService.send(message.getTo(), message.getTitle(), message.getContent());

            System.out.printf("✅ 邮件通知处理完成: MsgId=%s%n", msgId);
            return true;
        } catch (Exception e) {
            System.err.printf("❌ 处理邮件通知失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理短信通知
     */
    private boolean processSmsNotification(RocketMQProducerService.NotificationMessage message, String msgId) {
        try {
            // 模拟发送短信
            System.out.printf("📱 发送短信通知: 给%s%n", message.getTo());
            System.out.printf("💬 短信内容: %s%n", message.getContent());

            // 这里可以集成真实的短信发送服务
            // smsService.send(message.getTo(), message.getContent());

            System.out.printf("✅ 短信通知处理完成: MsgId=%s%n", msgId);
            return true;
        } catch (Exception e) {
            System.err.printf("❌ 处理短信通知失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    /**
     * 处理系统通知
     */
    private boolean processSystemNotification(RocketMQProducerService.NotificationMessage message, String msgId) {
        try {
            // 处理系统内部通知
            System.out.printf("⚙️ 处理系统通知: %s%n", message.getTitle());
            System.out.printf("📋 通知内容: %s%n", message.getContent());

            // 这里可以记录系统日志或触发其他系统操作

            System.out.printf("✅ 系统通知处理完成: MsgId=%s%n", msgId);
            return true;
        } catch (Exception e) {
            System.err.printf("❌ 处理系统通知失败: MsgId=%s, 错误=%s%n", msgId, e.getMessage());
            return false;
        }
    }

    // ========== 业务方法实现 ==========

    /**
     * 发送订单创建通知
     */
    private void sendOrderCreatedNotification(RocketMQProducerService.OrderCreatedMessage message) {
        try {
            String notificationContent = String.format(
                    "新订单已创建:\n" +
                            "订单ID: %d\n" +
                            "物流单号: %s\n" +
                            "收件人: %s\n" +
                            "物流公司: %s\n" +
                            "创建时间: %s",
                    message.getOrderId(),
                    message.getTrackingNumber(),
                    message.getRecipientName(),
                    message.getLogisticsCompany(),
                    new Date()
            );

            producerService.sendNotificationMessage(
                    "admin@example.com",
                    "新订单创建通知 - " + message.getTrackingNumber(),
                    notificationContent,
                    "email"
            );
        } catch (Exception e) {
            System.err.println("❌ 发送订单创建通知失败: " + e.getMessage());
        }
    }

    /**
     * 更新订单统计信息
     */
    private void updateOrderStatistics(RocketMQProducerService.OrderCreatedMessage message) {
        try {
            // 这里可以更新订单统计信息，如每日订单数等
            System.out.printf("📊 更新订单统计: 订单ID=%d%n", message.getOrderId());
        } catch (Exception e) {
            System.err.println("❌ 更新订单统计失败: " + e.getMessage());
        }
    }

    /**
     * 记录订单创建日志
     */
    private void logOrderCreation(RocketMQProducerService.OrderCreatedMessage message) {
        try {
            // 这里可以记录订单创建的操作日志
            System.out.printf("📝 记录订单创建日志: 订单ID=%d, 用户ID=%d%n",
                    message.getOrderId(), message.getUserId());
        } catch (Exception e) {
            System.err.println("❌ 记录订单创建日志失败: " + e.getMessage());
        }
    }

    /**
     * 发送状态变更通知
     */
    private void sendStatusChangeNotification(RocketMQProducerService.OrderStatusChangeMessage message) {
        try {
            String statusText = getStatusText(message.getNewStatus());
            String notificationContent = String.format(
                    "订单状态已更新:\n" +
                            "订单ID: %d\n" +
                            "物流单号: %s\n" +
                            "新状态: %s\n" +
                            "变更时间: %s",
                    message.getOrderId(),
                    message.getTrackingNumber(),
                    statusText,
                    new Date()
            );

            // 这里可以根据业务需要发送给相关用户或管理员
            producerService.sendNotificationMessage(
                    "user@example.com",
                    "订单状态更新通知 - " + message.getTrackingNumber(),
                    notificationContent,
                    "sms"
            );
        } catch (Exception e) {
            System.err.println("❌ 发送状态变更通知失败: " + e.getMessage());
        }
    }

    /**
     * 更新订单时间戳
     */
    private void updateOrderTimestamp(Long orderId) {
        try {
            // 这里可以更新订单的更新时间戳
            System.out.printf("🕒 更新订单时间戳: 订单ID=%d%n", orderId);
        } catch (Exception e) {
            System.err.println("❌ 更新订单时间戳失败: " + e.getMessage());
        }
    }

    /**
     * 发送签收成功通知
     */
    private void sendDeliverySuccessNotification(RocketMQProducerService.OrderDeliveredMessage message) {
        try {
            String notificationContent = String.format(
                    "订单已成功签收:\n" +
                            "订单ID: %d\n" +
                            "物流单号: %s\n" +
                            "签收时间: %s",
                    message.getOrderId(),
                    message.getTrackingNumber(),
                    new Date(message.getDeliveredTime())
            );

            producerService.sendNotificationMessage(
                    "user@example.com",
                    "订单签收成功 - " + message.getTrackingNumber(),
                    notificationContent,
                    "email"
            );
        } catch (Exception e) {
            System.err.println("❌ 发送签收成功通知失败: " + e.getMessage());
        }
    }

    /**
     * 标记订单为已完成
     */
    private void markOrderAsCompleted(Long orderId) {
        try {
            // 这里可以更新订单的完成状态
            System.out.printf("✅ 标记订单完成: 订单ID=%d%n", orderId);
        } catch (Exception e) {
            System.err.println("❌ 标记订单完成失败: " + e.getMessage());
        }
    }

    /**
     * 发送欢迎邮件
     */
    private void sendWelcomeEmail(RocketMQProducerService.UserRegisterMessage message) {
        try {
            String emailContent = String.format(
                    "亲爱的 %s，\n\n" +
                            "欢迎您注册我们的物流管理系统！\n\n" +
                            "您的账户信息：\n" +
                            "用户名：%s\n" +
                            "注册邮箱：%s\n\n" +
                            "感谢您的使用！",
                    message.getUsername(),
                    message.getUsername(),
                    message.getEmail()
            );

            producerService.sendNotificationMessage(
                    message.getEmail(),
                    "欢迎注册物流管理系统",
                    emailContent,
                    "email"
            );
        } catch (Exception e) {
            System.err.println("❌ 发送欢迎邮件失败: " + e.getMessage());
        }
    }

    /**
     * 发送新用户通知
     */
    private void sendNewUserNotification(RocketMQProducerService.UserRegisterMessage message) {
        try {
            String notificationContent = String.format(
                    "新用户注册:\n" +
                            "用户ID: %d\n" +
                            "用户名: %s\n" +
                            "邮箱: %s\n" +
                            "注册时间: %s",
                    message.getUserId(),
                    message.getUsername(),
                    message.getEmail(),
                    new Date(message.getRegisterTime())
            );

            producerService.sendNotificationMessage(
                    "admin@example.com",
                    "新用户注册通知",
                    notificationContent,
                    "system"
            );
        } catch (Exception e) {
            System.err.println("❌ 发送新用户通知失败: " + e.getMessage());
        }
    }

    /**
     * 初始化用户统计
     */
    private void initUserStatistics(Long userId) {
        try {
            // 这里可以初始化用户的统计信息
            System.out.printf("📈 初始化用户统计: 用户ID=%d%n", userId);
        } catch (Exception e) {
            System.err.println("❌ 初始化用户统计失败: " + e.getMessage());
        }
    }

    /**
     * 获取状态文本
     */
    private String getStatusText(Integer status) {
        if (status == null) return "未知";

        switch (status) {
            case LogisticsOrder.STATUS_PENDING: return "待发货";
            case LogisticsOrder.STATUS_SHIPPING: return "运输中";
            case LogisticsOrder.STATUS_DELIVERING: return "派送中";
            case LogisticsOrder.STATUS_DELIVERED: return "已签收";
            case LogisticsOrder.STATUS_EXCEPTION: return "异常";
            default: return "未知";
        }
    }

    /**
     * 获取订阅的主题信息
     */
    private String getSubscribedTopics() {
        return String.format("订单(%s), 用户(%s), 通知(%s)",
                MQConstants.TOPIC_ORDER,
                MQConstants.TOPIC_USER,
                MQConstants.TOPIC_NOTIFICATION);
    }

    /**
     * 检查消费者是否在运行
     */
    public boolean isRunning() {
        return running.get() && consumer != null;
    }

    /**
     * 重新启动消费者
     */
    public boolean restart() {
        System.out.println("🔄 尝试重新启动RocketMQ Consumer...");
        shutdown();
        initConsumer();
        return isRunning();
    }

    /**
     * 关闭消费者
     */
    public void shutdown() {
        if (consumer != null) {
            try {
                consumer.shutdown();
                running.set(false);
                System.out.println("✅ RocketMQ Consumer已关闭");
            } catch (Exception e) {
                System.err.println("❌ 关闭Consumer时出错: " + e.getMessage());
            } finally {
                consumer = null;
            }
        }
    }
}